/*
 * Copyright (c) 2020 - present, Inspur Genersoft Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.iec.edp.caf.rpc.client;

import com.google.protobuf.ByteString;
import io.grpc.Metadata;
import io.grpc.StatusRuntimeException;
//import io.iec.edp.caf.boot.context.CAFContext;
import io.iec.edp.caf.common.JSONSerializer;
import io.iec.edp.caf.commons.core.SerializerFactory;
import io.iec.edp.caf.commons.core.enums.SerializeType;
import io.iec.edp.caf.commons.exception.CAFRuntimeException;
import io.iec.edp.caf.commons.exception.ExceptionLevel;
import io.iec.edp.caf.commons.exception.entity.DefaultExceptionProperties;
import io.iec.edp.caf.commons.exception.entity.ExceptionErrorCode;
import io.iec.edp.caf.commons.runtime.CafEnvironment;
import io.iec.edp.caf.commons.runtime.thread.CAFThreadPoolExecutor;
//import io.iec.edp.caf.core.session.ICafSessionService;
import io.iec.edp.caf.logging.CommonConstant;
import io.iec.edp.caf.msu.client.exception.ServiceUnitNotFoundException;
import io.iec.edp.caf.rpc.api.channel.RpcAbstractChannel;
import io.iec.edp.caf.rpc.api.channel.RpcChanelAsynCallBack;
import io.iec.edp.caf.rpc.api.common.RpcChannelType;
import io.iec.edp.caf.rpc.api.entity.RpcServiceMethodDefinition;
import io.iec.edp.caf.rpc.api.event.RpcClientEventBroker;
import io.iec.edp.caf.rpc.api.grpc.GrpcVariable;
import io.iec.edp.caf.rpc.api.serialize.RpcSerializeUtil;
import io.iec.edp.caf.rpc.api.service.CAFRpcClient;
import io.iec.edp.caf.rpc.api.service.InternalServiceManageService;
import io.iec.edp.caf.rpc.api.support.*;
import io.iec.edp.caf.rpc.api.utils.Utils;
import io.iec.edp.caf.rpc.api.utils.Validator;
import io.iec.edp.caf.rpc.client.local.RpcLocalChecker;
import io.iec.edp.caf.rpc.client.local.RpcLocalInvoker;
//import io.iec.edp.caf.tenancy.api.ITenantRouteService;
//import io.iec.edp.caf.tenancy.api.ITenantService;
//import io.iec.edp.caf.tenancy.api.context.RequestTenantContextHolder;
//import io.iec.edp.caf.tenancy.api.context.RequestTenantContextInfo;
//import io.iec.edp.caf.tenancy.api.entity.TenancyMode;
//import io.iec.edp.caf.tenancy.api.exception.TenantNotFoundException;
import io.iec.edp.caf.rpc.api.spi.RpcContextResolver;
import io.iec.edp.caf.rpc.api.spi.RpcContextTransmittee;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.http.Cookie;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationTargetException;
import java.nio.charset.StandardCharsets;
import java.time.OffsetDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.stream.Collectors.toList;

@Slf4j
public class CAFRpcClientImpl implements CAFRpcClient {

    private final InternalServiceManageService management;
    private final RpcLocalInvoker rpcLocalInvoker;
//    private final ITenantService tenantService;
//    private final ITenantRouteService tenantRouteService;
    private String cookieName;
    private final RpcClientEventBroker clientEventBroker;

    private static final String HTTP_THREAD_NAME="http-nio-";
    private static final String RPC_CLIENT_THREAD_NAME="RPC-Client-LocalInvoke-Thread-";
    private static final ThreadPoolExecutor executor =
            //核心数和最大数默认给16和64，但是提供可配置的方式。队列数默认为1，不让任务等待，暂不提供配置
            //构造该类时会根据cpu逻辑核调整corepoolsize和maxpoolsize
            new CAFThreadPoolExecutor(64, 200, 10, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<Runnable>(1),
                    new ThreadFactory() {
                        private AtomicInteger id = new AtomicInteger(0);
                        @Override
                        public Thread newThread(Runnable r) {
                            Thread thread = new Thread(r);
                            thread.setName(RPC_CLIENT_THREAD_NAME + id.addAndGet(1));
                            return thread;
                        }
                    }, new ThreadPoolExecutor.AbortPolicy());

    public CAFRpcClientImpl(RpcClientEventBroker clientEventBroker,
                            InternalServiceManageService management,
                            RpcLocalInvoker rpcLocalInvoker
//                          ITenantService tenantService,
//                          ITenantRouteService tenantRouteService
    ) {

        this.clientEventBroker = clientEventBroker;
        this.management = management;
        this.rpcLocalInvoker = rpcLocalInvoker;
        this.cookieName = CafEnvironment.getEnvironment().getProperty("caf-security.general.cookie-name");
//        this.tenantService = tenantService;
//        this.tenantRouteService = tenantRouteService;
        if (this.cookieName == null) {
            this.cookieName = ConstanceVarible.COOKIE_SESSION_NAME;
        }
    }

    @SneakyThrows
    @Override
    public <T> T invoke(Class<T> t, String serviceId, String serviceUnitName, LinkedHashMap<String, Object> parameters, RpcChannelType channelType, HashMap<String, String> context) {
        return this.invoke(new Type<>(t),serviceId,"latest",serviceUnitName, parameters,channelType,ConstanceVarible.RPC_FILTER_MODE_DEFAULT,context);
    }

    @SneakyThrows
    @Override
    public <T> T invoke(Class<T> t, String serviceId, String version,String serviceUnitName,LinkedHashMap<String, Object> parameters, RpcChannelType channelType,String filterType, HashMap<String, String> context) {
        return this.invoke(new Type<>(t), serviceId, version,serviceUnitName,parameters, channelType,filterType,context);
    }

    @Override
    public <T> T invoke(Type<T> t, String serviceId, String serviceUnitName, LinkedHashMap<String, Object> parameters, RpcChannelType channelType, HashMap<String, String> context) throws IllegalAccessException, InstantiationException, IOException, NoSuchMethodException, InvocationTargetException, ClassNotFoundException {
        return this.invoke(t, serviceId, "latest", serviceUnitName, parameters, channelType,ConstanceVarible.RPC_FILTER_MODE_DEFAULT,context);
    }

    @Override
    public String invoke(String serviceId,String version, String serviceUnitName, LinkedHashMap<String, Object> parameters, RpcChannelType channelType,String filterType, HashMap<String, String> context) {
        return this.invoke(String.class, serviceId, version, serviceUnitName, parameters, channelType,filterType,context);
    }


    @SneakyThrows
    @Override
    public <T> T invoke(Type<T> t, String serviceId,String version, String serviceUnitName, LinkedHashMap<String, Object> parameters, RpcChannelType channelType, String filterType,HashMap<String, String> context) throws IllegalAccessException, InstantiationException, IOException, NoSuchMethodException, InvocationTargetException, ClassNotFoundException {
        if(version==null || version.length()==0){
            version = "v1.0";
        }

        if(filterType==null || filterType.length() ==0){
            filterType = ConstanceVarible.RPC_FILTER_MODE_DEFAULT;
        }

        //兼容RPC1.0接口，后续可以基于RPC2.0的REST接口实现
//        if(channelType!=null&&channelType.equals(RpcChannelType.REST)){
//            RpcClient rpcClient = SpringBeanUtils.getBean(RpcClient.class);
//            assert rpcClient != null;
//            return rpcClient.invoke(t,serviceId,serviceUnitName,parameters,context);
//        }

        try {
            var start = System.currentTimeMillis();
            var end =  System.currentTimeMillis();

            //get cookies
            Cookie[] cks = this.getCookies();
            //check cookie only log warning not break invoke
            Validator.CheckCksSession(cks,serviceId);

            T result;
            //get rpc service definition from local memory variable
            RpcServiceMethodDefinition serviceMethodDefinition = this.management.getRpcMethodDefinition(serviceId);
            //get extend tenant id
            RpcContextResolver resolver = RpcContextResolver.getRpcContextResolver();
            Integer targetTenantId = resolver!=null?resolver.resolveRequestTargetTenantId(serviceUnitName, context):null;

            //rpc service event context
            HashMap<String, String> eventContext = new HashMap<>();
            eventContext.put(ConstanceVarible.CURRENT_SERVICE_ID, serviceId);
            eventContext.put(ConstanceVarible.TARGET_SU,serviceUnitName);

            log.info("start rpc invoke，current serviceId：{}，current target su:{}",serviceId,serviceUnitName);
            Object retvalue = null;

            //is local invoke
            if (RpcLocalChecker.isLocalInvoke(serviceUnitName)) {
                //local memory has no rpc service definition
                if (serviceMethodDefinition == null) {
                    log.error("can not find rpc service by serviceid:"+serviceId+"，can not local invoke,current su："+serviceUnitName);
                    throw new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                            DefaultExceptionProperties.RESOURCE_FILE,
                            ExceptionErrorCode.localServiceUnitNotFound,
                            new String[]{serviceId, serviceUnitName},
                            null, ExceptionLevel.Error, false);
                }

                eventContext.put(ConstanceVarible.IS_LOCAL, "true");

                if(log.isInfoEnabled()){
                    log.info("rpc local invoke info：", JSONSerializer.serialize(serviceMethodDefinition));
                }

                //local invoke（bean first,then reflect）
                start = System.currentTimeMillis();
                //rpcLocalInvoker has rpc event invoke
                result = this.rpcLocalInvoker.invokeLocalService(t.getRawType(), serviceId, serviceUnitName, parameters, serviceMethodDefinition, targetTenantId,eventContext);
                end =  System.currentTimeMillis();
                log.info("rpc local invoke client total time：{} ms",end-start);
                retvalue = result;

            } else {
                channelType = channelType==null?RpcChannelType.GRPC:channelType;
                filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType;
                //create rpc channel by protocol
                var channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext);
                assert channel != null;
                if(log.isInfoEnabled()){
                    log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort());
                }

                //prepare event args
                eventContext.put(ConstanceVarible.IS_LOCAL, "false");
                eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort());
                eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName());
                HashMap<String, Object> localDict = new HashMap<>();
                localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName);
                //before invoke event
                this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict);
                String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID);

                //set gsp context to header
                HashMap<String, String> gspContext = new HashMap<>();
                setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext);

                //add additional headers to channel
                Map<String,String> headers = new HashMap<>();
                headers.put(CommonConstant.TRACE_ID_HEADER,traceId);
                headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext));
                channel.addHeaders(headers);

                //in RPC2.0 use the protobuf serialize as default
                LinkedHashMap<String, String> params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition);
//                LinkedHashMap<String, String> params = RpcSerializeUtil.serializeParameter(parameters, serviceMethodDefinition);

                start = System.currentTimeMillis();
                var returnString = channel.invokeRemoteService(serviceId,version,params,null);
                channel.stopClient();
//                retvalue = RpcSerializeUtil.deSerializeReturnValue(t, returnString ,serviceMethodDefinition);
                retvalue = RpcSerializeUtil.deSerializeReturnValue4RPC2(t, returnString ,serviceMethodDefinition);

                var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT);
                var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class);
                serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict;
                this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict);

                end =  System.currentTimeMillis();

                if(log.isInfoEnabled()){
                    log.info("rpc remote invoke client total time:{} ms",end-start);
                }

            }

            return (T)retvalue;
        } catch (Exception e) {
            log.debug("rpc client exception type:{}",e.getClass().getName());
            var ex = handleException(e);
            throw ex;
        }finally {
            RpcThreadCacheHolder.clear();
        }
    }


    @SneakyThrows
    @Override
    public InputStream invokeStream(String serviceId, String version, String serviceUnitName, LinkedHashMap<String, Object> parameters, RpcChannelType channelType, String filterType, HashMap<String, String> context) {
        if(version==null || version.length()==0){
            version = "v1.0";
        }

        if(filterType==null || filterType.length() ==0){
            filterType = ConstanceVarible.RPC_FILTER_MODE_DEFAULT;
        }

        //兼容RPC1.0接口，后续可以基于RPC2.0的REST接口实现
//        if(channelType!=null&&channelType.equals(RpcChannelType.REST)){
//            RpcClient rpcClient = SpringBeanUtils.getBean(RpcClient.class);
//            assert rpcClient != null;
//            return rpcClient.invokeStream(serviceId,version,serviceUnitName,parameters,context);
//        }

        try {

            var start = System.currentTimeMillis();
            var end =  System.currentTimeMillis();

            //get cookies
            Cookie[] cks = this.getCookies();
            //check cookie only log warning not break invoke
            Validator.CheckCksSession(cks,serviceId);

            //get rpc service definition from local memory variable
            RpcServiceMethodDefinition serviceMethodDefinition = this.management.getRpcMethodDefinition(serviceId);
            //get extend tenant id
            RpcContextResolver resolver = RpcContextResolver.getRpcContextResolver();
            Integer targetTenantId = resolver!=null?resolver.resolveRequestTargetTenantId(serviceUnitName, context):null;

            //rpc service event context
            HashMap<String, String> eventContext = new HashMap<>();
            eventContext.put(ConstanceVarible.CURRENT_SERVICE_ID, serviceId);
            eventContext.put(ConstanceVarible.TARGET_SU,serviceUnitName);

            log.info("start rpc invoke，current serviceId：{}，current target su:{}",serviceId,serviceUnitName);
            InputStream retvalue = null;

            //is local invoke
            if (RpcLocalChecker.isLocalInvoke(serviceUnitName)) {
                //local memory has no rpc service definition
                if (serviceMethodDefinition == null) {
                    log.error("can not find rpc service by serviceid:"+serviceId+"，can not local invoke,current su："+serviceUnitName);
                    throw new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                            DefaultExceptionProperties.RESOURCE_FILE,
                            ExceptionErrorCode.localServiceUnitNotFound,
                            new String[]{serviceId, serviceUnitName},
                            null, ExceptionLevel.Error, false);
                }

                eventContext.put(ConstanceVarible.IS_LOCAL, "true");

                if(log.isInfoEnabled()){
                    log.info("rpc local invoke info：", JSONSerializer.serialize(serviceMethodDefinition));
                }

                //local invoke（bean first,then reflect）
                start = System.currentTimeMillis();
                retvalue = this.rpcLocalInvoker.invokeLocalServiceStream(serviceId,version, serviceUnitName, parameters, serviceMethodDefinition, targetTenantId,eventContext);
                end =  System.currentTimeMillis();
                log.info("rpc local invoke client total time：{} ms",end-start);
            } else {
                channelType = channelType==null?RpcChannelType.GRPC:channelType;
                filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType;
                //create rpc channel by protocol
                var channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext);
                assert channel != null;
                if(log.isInfoEnabled()){
                    log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort());
                }

                //prepare event args
                eventContext.put(ConstanceVarible.IS_LOCAL, "false");
                eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort());
                eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName());
                HashMap<String, Object> localDict = new HashMap<>();
                localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName);
                //before invoke event
                this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict);
                String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID);

                //prepare rpc context
                HashMap<String, String> gspContext = new HashMap<>();
                setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext);

                //add additional headers to channel
                Map<String,String> headers = new HashMap<>();
                headers.put(CommonConstant.TRACE_ID_HEADER,traceId);
                headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext));
                channel.addHeaders(headers);

                //in RPC2.0 use the protobuf serialize as default
                LinkedHashMap<String, String> params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition);
//                LinkedHashMap<String, String> params = RpcSerializeUtil.serializeParameter(parameters, serviceMethodDefinition);

                start = System.currentTimeMillis();
                retvalue = channel.invokeRemoteServiceStream(serviceId,version,params,null);
                channel.stopClient();

                var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT);
                var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class);
                serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict;
                this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict);

                end =  System.currentTimeMillis();

                if(log.isInfoEnabled()){
                    log.info("rpc remote invoke client total time:{} ms",end-start);
                }

            }

            return retvalue;
        } catch (Exception e) {
            log.debug("rpc client exception type:{}",e.getClass().getName());
            var ex = handleException(e);
            throw ex;
        }finally {
            RpcThreadCacheHolder.clear();
        }
    }

    @SneakyThrows
    @Override
    public void invokeStreamAsyn(String serviceId, String version, String serviceUnitName, LinkedHashMap<String, Object> parameters, RpcChannelType channelType, String filterType, HashMap<String, String> context, RpcChanelAsynCallBack callBack) {
        if(version==null || version.length()==0){
            version = "v1.0";
        }

        if(filterType==null || filterType.length() ==0){
            filterType = ConstanceVarible.RPC_FILTER_MODE_DEFAULT;
        }

        //兼容RPC1.0接口，后续可以基于RPC2.0的REST接口实现
//        if(channelType!=null&&channelType.equals(RpcChannelType.REST)){
//            RpcClient rpcClient = SpringBeanUtils.getBean(RpcClient.class);
//            assert rpcClient != null;
//            return rpcClient.invokeStream(serviceId,version,serviceUnitName,parameters,context);
//        }

        try {

            var start = System.currentTimeMillis();
            var end =  System.currentTimeMillis();

            //get cookies
            Cookie[] cks = this.getCookies();
            //check cookie only log warning not break invoke
            Validator.CheckCksSession(cks,serviceId);

            //get rpc service definition from local memory variable
            RpcServiceMethodDefinition serviceMethodDefinition = this.management.getRpcMethodDefinition(serviceId);
            //get extend tenant id
            RpcContextResolver resolver = RpcContextResolver.getRpcContextResolver();
            Integer targetTenantId = resolver!=null?resolver.resolveRequestTargetTenantId(serviceUnitName, context):null;

            //rpc service event context
            HashMap<String, String> eventContext = new HashMap<>();
            eventContext.put(ConstanceVarible.CURRENT_SERVICE_ID, serviceId);
            eventContext.put(ConstanceVarible.TARGET_SU,serviceUnitName);

            log.info("start rpc invoke，current serviceId：{}，current target su:{}",serviceId,serviceUnitName);

            //is local invoke
            if (RpcLocalChecker.isLocalInvoke(serviceUnitName)) {
                //local memory has no rpc service definition
                if (serviceMethodDefinition == null) {
                    log.error("can not find rpc service by serviceid:"+serviceId+"，can not local invoke,current su："+serviceUnitName);
                    throw new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                            DefaultExceptionProperties.RESOURCE_FILE,
                            ExceptionErrorCode.localServiceUnitNotFound,
                            new String[]{serviceId, serviceUnitName},
                            null, ExceptionLevel.Error, false);
                }

                eventContext.put(ConstanceVarible.IS_LOCAL, "true");

                if(log.isInfoEnabled()){
                    log.info("rpc local invoke info：", JSONSerializer.serialize(serviceMethodDefinition));
                }

                //local invoke（bean first,then reflect）
                start = System.currentTimeMillis();
                var ipsValue = this.rpcLocalInvoker.invokeLocalServiceStream(serviceId,version, serviceUnitName, parameters, serviceMethodDefinition, targetTenantId,eventContext);
                if(Thread.currentThread().getName().startsWith(HTTP_THREAD_NAME)){
                    executor.execute(new DataAsynHandler(ipsValue,callBack));
                }else{
                    Thread thread = new Thread(new DataAsynHandler(ipsValue,callBack));
                    thread.start();
                }
                end =  System.currentTimeMillis();
                log.info("rpc local invoke client total time：{} ms",end-start);
            } else {
                channelType = channelType==null?RpcChannelType.GRPC:channelType;
                filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType;
                //create rpc channel by protocol
                var channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext);
                assert channel != null;
                if(log.isInfoEnabled()){
                    log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort());
                }

                //prepare event args
                eventContext.put(ConstanceVarible.IS_LOCAL, "false");
                eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort());
                eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName());
                HashMap<String, Object> localDict = new HashMap<>();
                localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName);
                //before invoke event
                this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict);
                String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID);

                //prepare rpc context
                HashMap<String, String> gspContext = new HashMap<>();
                setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext);

                //add additional headers to channel
                Map<String,String> headers = new HashMap<>();
                headers.put(CommonConstant.TRACE_ID_HEADER,traceId);
                headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext));
                channel.addHeaders(headers);

                //in RPC2.0 use the protobuf serialize as default
                LinkedHashMap<String, String> params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition);
//                LinkedHashMap<String, String> params = RpcSerializeUtil.serializeParameter(parameters, serviceMethodDefinition);

                start = System.currentTimeMillis();
                channel.invokeRemoteServiceStreamAsyn(serviceId,version,params,null,callBack);
                channel.stopClient();

                var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT);
                var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class);
                serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict;
                this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, null, localDict);

                end =  System.currentTimeMillis();

                if(log.isInfoEnabled()){
                    log.info("rpc remote invoke client total time:{} ms",end-start);
                }

            }

        } catch (Exception e) {
            log.debug("rpc client exception type:{}",e.getClass().getName());
            var ex = handleException(e);
            throw ex;
        }finally {
            RpcThreadCacheHolder.clear();
        }
    }

    @SneakyThrows
    @Override
    public <T> T invokeStream(Type<T> t, String serviceId, String version, String serviceUnitName, LinkedHashMap<String, Object> parameters, RpcChannelType channelType,String filterType, HashMap<String, String> context) {
        if(version==null || version.length()==0){
            version = "v1.0";
        }

        if(filterType==null || filterType.length() ==0){
            filterType = ConstanceVarible.RPC_FILTER_MODE_DEFAULT;
        }

        //兼容RPC1.0接口，后续可以基于RPC2.0的REST接口实现
//        if(channelType!=null&&channelType.equals(RpcChannelType.REST)){
//            RpcClient rpcClient = SpringBeanUtils.getBean(RpcClient.class);
//            assert rpcClient != null;
//            return rpcClient.invokeStream(stream,t,serviceId,version,serviceUnitName,parameters,context);
//        }

        try {

            var start = System.currentTimeMillis();
            var end =  System.currentTimeMillis();

            //get cookies
            Cookie[] cks = this.getCookies();
            //check cookie only log warning not break invoke
            Validator.CheckCksSession(cks,serviceId);

//            T result;
            //get rpc service definition from local memory variable
            RpcServiceMethodDefinition serviceMethodDefinition = this.management.getRpcMethodDefinition(serviceId);
            //get extend tenant id
            RpcContextResolver resolver = RpcContextResolver.getRpcContextResolver();
            Integer targetTenantId = resolver!=null?resolver.resolveRequestTargetTenantId(serviceUnitName, context):null;

            //rpc service event context
            HashMap<String, String> eventContext = new HashMap<>();
            eventContext.put(ConstanceVarible.CURRENT_SERVICE_ID, serviceId);
            eventContext.put(ConstanceVarible.TARGET_SU,serviceUnitName);

            log.info("start rpc invoke，current serviceId：{}，current target su:{}",serviceId,serviceUnitName);
            Object retvalue = null;

            //is local invoke
            if (RpcLocalChecker.isLocalInvoke(serviceUnitName)) {
                //local memory has no rpc service definition
                if (serviceMethodDefinition == null) {
                    log.error("can not find rpc service by serviceid:"+serviceId+"，can not local invoke,current su："+serviceUnitName);
                    throw new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                            DefaultExceptionProperties.RESOURCE_FILE,
                            ExceptionErrorCode.localServiceUnitNotFound,
                            new String[]{serviceId, serviceUnitName},
                            null, ExceptionLevel.Error, false);
                }

                eventContext.put(ConstanceVarible.IS_LOCAL, "true");

                if(log.isInfoEnabled()){
                    log.info("rpc local invoke info：", JSONSerializer.serialize(serviceMethodDefinition));
                }

                //local invoke（bean first,then reflect）
                start = System.currentTimeMillis();
                var stream = Utils.handleStreamFromParameters(parameters);
                retvalue = this.rpcLocalInvoker.invokeLocalServiceStream(stream,t.getRawType(), serviceId,version, serviceUnitName, parameters, serviceMethodDefinition, targetTenantId,eventContext);
                end =  System.currentTimeMillis();
                log.info("rpc local invoke client total time：{} ms",end-start);

            } else {
                channelType = channelType==null?RpcChannelType.GRPC:channelType;
                filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType;
                //create rpc channel by protocol
                var channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext);
                assert channel != null;
                if(log.isInfoEnabled()){
                    log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort());
                }

                //prepare event args
                eventContext.put(ConstanceVarible.IS_LOCAL, "false");
                eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort());
                eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName());
                HashMap<String, Object> localDict = new HashMap<>();
                localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName);
                //before invoke event
                this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict);
                String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID);

                //prepare rpc context
                HashMap<String, String> gspContext = new HashMap<>();
                setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext);

                //add additional headers to channel
                Map<String,String> headers = new HashMap<>();
                headers.put(CommonConstant.TRACE_ID_HEADER,traceId);
                headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext));
                channel.addHeaders(headers);

                //here parse the inputstream parameter in parameters
                var stream = Utils.handleStreamFromParameters(parameters);
                //in RPC2.0 use the protobuf serialize as default
                LinkedHashMap<String, String> params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition);
//                LinkedHashMap<String, String> params = RpcSerializeUtil.serializeParameter(parameters, serviceMethodDefinition);

                start = System.currentTimeMillis();

                var returnString = channel.invokeRemoteServiceStream(stream,serviceId,version,params,null);
                channel.stopClient();
//                retvalue = RpcSerializeUtil.deSerializeReturnValue(t, returnString ,serviceMethodDefinition);
                retvalue = RpcSerializeUtil.deSerializeReturnValue4RPC2(t, returnString ,serviceMethodDefinition);

                var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT);
                var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class);
                serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict;
                this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict);

                end =  System.currentTimeMillis();

                if(log.isInfoEnabled()){
                    log.info("rpc remote invoke client total time:{} ms",end-start);
                }

            }

            return (T)retvalue;
        } catch (Exception e) {
            log.debug("rpc client exception type:{}",e.getClass().getName());
            var ex = handleException(e);
            throw ex;
        }finally {
            RpcThreadCacheHolder.clear();
        }
    }

    @Override
    public void setTimeout(Integer timeout) {
        RpcTimeoutHolder.setTimeout(timeout);
    }

    /**
     * get cookie
     * @return if the request has cookie and has caf session,then use the request cookie
     *         if request has no cookie then use {@link CAFContext} session create a new cookie
     * @throws UnsupportedEncodingException
     */
    private Cookie[] getCookies() throws UnsupportedEncodingException {
        Cookie[] cks = null;
        boolean containSessionId = false;
        //get request cookies
        ServletRequestAttributes servletRequestAttributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        var sessionid = "";
        if (servletRequestAttributes != null) {
            HttpServletRequest request = servletRequestAttributes.getRequest();
            cks = request.getCookies();
            //if cookie has session,then use cookie session
            if(cks!=null && cks.length>0){
                for (Cookie cookie: cks) {
                    if(cookie.getName().equalsIgnoreCase(this.cookieName)){
                        containSessionId = true;
                        sessionid = cookie.getValue();
                        break;
                    }
                }
            }
        }

        var context = RpcContextResolver.capture(RpcContextTransmittee.TransmitteType.SESSION.getValue());
        if(!containSessionId){
            log.debug("current cookies has no sessionid,get sessionid from CAFContext");
            cks = new Cookie[1];
            //here getContext key must match SessionTransmitter`s key
            cks[0] = new Cookie(cookieName, base64Encode((String) context.getContext("transmitte-cafsessionid")));
        }else{
            log.debug("current cookies has sessionid no decode:"+sessionid);
            log.debug("current cookies has sessionid decode sessionid:"+new String(Base64.getDecoder().decode(sessionid), StandardCharsets.UTF_8));
            context.putContext("transmitte-cafsessionid",sessionid);
        }

        RpcContextResolver.getRpcContextResolver().validateRpcContext(context);
        return cks;
    }

    private String base64Encode(String value) throws UnsupportedEncodingException {

        if (org.springframework.util.StringUtils.isEmpty(value)) {
            return "";
        }

        byte[] encodedCookieBytes = Base64.getEncoder().encode(value.getBytes("UTF-8"));
        return new String(encodedCookieBytes);
    }

    private void setContext(Map<String,String> gspContext, Cookie[] cks, String serviceUnitName, Integer tenantId, Map<String,String> eventContext){
        String sessionId = "";
        //set some context for early net core
        if (cks != null && cks.length > 0) {
            List<Cookie> ck = Arrays.stream(cks).filter(c -> c.getName().equalsIgnoreCase(ConstanceVarible.NET_SESSION_ID)).collect(toList());
            sessionId = (ck != null && ck.size() > 0) ? ck.get(0).getValue() : sessionId;
        }
        gspContext.put(ConstanceVarible.GSP_CONTEXT_ID, sessionId);
        gspContext.put(ConstanceVarible.GSP_RPC, "true");
        gspContext.put(ConstanceVarible.GSP_MSU, serviceUnitName);
        if (tenantId != null) {
            gspContext.put(ConstanceVarible.GSP_RPC_TENANT, tenantId.toString());
        }
        gspContext.put(ConstanceVarible.GSP_RPC_CLIENT_ENVENT, SerializerFactory.getSerializer(SerializeType.Json).serializeToString(eventContext));

    }


//    /**
//     * get extend tenantid to route
//     * by implement {@link ITenantRouteService} interface
//     * @param msu msu code
//     * @param context extend parameters to choose route tenant
//     * @return if {@link ITenantRouteService} implement it will reture the ITenantRouteService`s tenantid,if not implement
//     *          {@link ITenantRouteService} it will found tenantid in http header by CAFContextFilter,if both not found it
//     *          will return null
//     */
//    private Integer getTargetTenantId(String msu, HashMap<String, String> context) {
//
//        if (this.tenantService.getTenancyMode() != TenancyMode.group)
//            return null;
//
//        context = context == null ? new HashMap<>() : context;
//        String tenantIdString = context.get(ConstanceVarible.ROUTE_TENANTID);
//        if (tenantIdString != null) {
//            return Integer.valueOf(tenantIdString);
//        } else {
//            String tenantDim1 = context.get(ConstanceVarible.ROUTE_FIRSTDIMENSION_KEY);
//            String tenantDim2 = context.get(ConstanceVarible.ROUTE_SECONDDIMENSION_KEY);
//            if (!StringUtils.isEmpty(tenantDim1)) {
//                try {
//                    return tenantRouteService.route(msu, tenantDim1, tenantDim2);
//                } catch (TenantNotFoundException e) {
//                    log.error(String.format("tenant route failed su：%s first dimension：%s ,second dimension：%s", msu, tenantDim1, tenantDim2));
//                }
//            }
//
//            //get requset header tenant if set
//            RequestTenantContextInfo contextInfo = RequestTenantContextHolder.get();
//            return contextInfo == null ? null : contextInfo.getTenantId();
//        }
//    }

    /**
     * https://javakk.com/2558.html
     * use metadata to deal with exception
     */
    @SneakyThrows
    private Throwable handleException(Throwable e){

        if(e instanceof ServiceUnitNotFoundException){
            return e;
        }else if(e instanceof StatusRuntimeException){
            Metadata trailers = ((StatusRuntimeException)e).getTrailers();
            var serializer = SerializerFactory.getSerializer(SerializeType.Protobuf);
            String code = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_CODE, Metadata.ASCII_STRING_MARSHALLER));
            int level = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER))==null?
                    2:serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER)),Integer.class);
            String message = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER));
            String innerMessage = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER));
            OffsetDateTime date =trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER))==null?
                    OffsetDateTime.now():serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER)),OffsetDateTime.class);
            String detail = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER))==null?
                    "":serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER)),String.class);
            String requestId = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_REQUESTID, Metadata.ASCII_STRING_MARSHALLER));
            String exceptionType = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_TYPE, Metadata.ASCII_STRING_MARSHALLER));
            boolean isBizException=trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER))==null?
                        false:serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER)),Boolean.class);;
            HashMap<String, String> extensionMessage = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER))==null?
                    new HashMap<>():serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)),HashMap.class);
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_CODE, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                code = ;
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                level = ;
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                message = ;
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                innerMessage = ;
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                var temp = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER));
//                if(temp!=null){
//                    date = serializer.deserialize(temp,OffsetDateTime.class);
//                }
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                detail = ;
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_REQUESTID, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                requestId = ;
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_TYPE, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                exceptionType = ;
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER))!=null){
//                var temp = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER));
//                if(temp!=null){
//                    isBizException = serializer.deserialize(temp,Boolean.class);
//                }
//            }
//            if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER))){
//                var temp = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER));
//                if(temp!=null){
//                    extensionMessage = serializer.deserialize(temp,HashMap.class);
//                }
//            }


            if (exceptionType != null && exceptionType.equalsIgnoreCase(CAFRuntimeException.class.getTypeName())) {
                return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, code, message + "\n" + detail, null, ExceptionLevel.getLevel(level), isBizException);
            } else {
                return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                        ExceptionErrorCode.rpcError, message + "\n" + detail, null, ExceptionLevel.Error, false);
            }
        }else{
            return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT,
                    DefaultExceptionProperties.RESOURCE_FILE,
                    ExceptionErrorCode.rpcError,
                    new String[]{}, e, ExceptionLevel.Error, false);
        }
    }

    class DataAsynHandler implements Runnable{
        private InputStream ipsValue;
        private RpcChanelAsynCallBack callBack;
        public DataAsynHandler(InputStream ipsValue,RpcChanelAsynCallBack callBack){
            this.ipsValue = ipsValue;
            this.callBack = callBack;
        }
        @SneakyThrows
        @Override
        public void run() {
            try{
                // upload file as chunk
                byte[] bytes = new byte[4096];
                int size;
                while ((size = ipsValue.read(bytes)) > 0){
                    //build request parameter
                    var data = ByteString.copyFrom(bytes, 0 , size);
                    callBack.onCallBack(data);
                }
            }catch (Exception e){
                log.error(e.getMessage(),e);
            }finally {
                if (ipsValue != null) {
                    try {
                        ipsValue.close();
                    } catch (IOException e) {
                        log.error(e.getMessage(),e);
                    }
                }
            }
        }
    }
}
